package com.kk.xhr.admin.config;

import com.kk.xhr.config.RejectedPolicy;
import com.kk.xhr.config.ThreadPoolType;
import com.kk.xhr.config.porperties.ExecutorProperties;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.PropertySource;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import java.util.concurrent.Executor;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.ForkJoinWorkerThread;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 线程池配置
 *
 * @author luokexiong
 * @version 1.0 2021/2/15
 * @since 1.0.0
 */
@Configuration
@PropertySource(value = {
    "classpath:executor.properties"
})
@EnableAsync(proxyTargetClass = true)
@Slf4j
public class ExecutorConfig {

    @Bean
    @ConfigurationProperties(prefix = "executor.admin.biz")
    public ExecutorProperties adminControllerBizProperties() {
        return new ExecutorProperties();
    }

    @Bean
    @ConfigurationProperties(prefix = "executor.file")
    public ExecutorProperties fileControllerProperties() {
        return new ExecutorProperties();
    }

    @Bean
    @ConfigurationProperties(prefix = "executor.message")
    public ExecutorProperties messageProperties() {
        return new ExecutorProperties();
    }

    /**
     * 管理端异步Controller线程池
     *
     * @return 线程池
     */
    @Bean(value = "adminControllerBizExecutor", destroyMethod = "shutdown")
    public Executor adminControllerBizExecutor() {
        ExecutorProperties properties = adminControllerBizProperties();
        if (properties.getPoolType() == ThreadPoolType.FORKJOIN) {
            return createForkJoinExecutor(properties);
        }
        return createNormalExecutor(properties);
    }

    /**
     * 文件异步Controller线程池
     *
     * @return 线程池
     */
    @Bean(value = "fileControllerExecutor", destroyMethod = "shutdown")
    public Executor fileControllerExecutor() {
        ExecutorProperties properties = fileControllerProperties();
        if (properties.getPoolType() == ThreadPoolType.FORKJOIN) {
            return createForkJoinExecutor(properties);
        }
        return createNormalExecutor(properties);
    }

    /**
     * 文件异步Controller线程池
     *
     * @return 线程池
     */
    @Bean(value = "messageExecutor", destroyMethod = "shutdown")
    public Executor messageExecutor() {
        ExecutorProperties properties = messageProperties();
        if (properties.getPoolType() == ThreadPoolType.FORKJOIN) {
            return createForkJoinExecutor(properties);
        }
        return createNormalExecutor(properties);
    }

    /**
     * 创建本地线程池
     *
     * @param properties 配置类
     * @return 线程池
     */
    private Executor createForkJoinExecutor(ExecutorProperties properties) {
        int parallelism = null == properties.getParallelism() ? Runtime.getRuntime().availableProcessors() : properties.getParallelism();
        boolean asyncMode = null == properties.getAsyncMode() || properties.getAsyncMode();
        ForkJoinPool.ForkJoinWorkerThreadFactory factory = new NamedForkJoinWorkerThreadFactory(properties.getThreadNamePrefix());
        return new ForkJoinPool(parallelism, factory, null, asyncMode);
    }

    /**
     * 创建普通线程池
     *
     * @param properties 配置类
     * @return 线程池
     */
    private Executor createNormalExecutor(ExecutorProperties properties) {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(properties.getCorePoolSize());
        executor.setAllowCoreThreadTimeOut(properties.getAllowCoreThreadTimeout());
        executor.setMaxPoolSize(properties.getMaxPoolSize());
        executor.setQueueCapacity(properties.getQueueCapacity());
        executor.setThreadNamePrefix(properties.getThreadNamePrefix());
        executor.setRejectedExecutionHandler(RejectedPolicy.parseRejectedHandler(properties.getRejectedPolicy()));
        executor.initialize();
        return executor;
    }

    /**
     * 带命名规则的ForkJoinWorkerThreadFactory
     */
    public static class NamedForkJoinWorkerThreadFactory implements ForkJoinPool.ForkJoinWorkerThreadFactory {

        final String prefix;
        final AtomicInteger count = new AtomicInteger();

        public NamedForkJoinWorkerThreadFactory(String prefix) {
            this.prefix = prefix;
        }

        @Override
        public ForkJoinWorkerThread newThread(ForkJoinPool pool) {
            ForkJoinWorkerThread thread = ForkJoinPool.defaultForkJoinWorkerThreadFactory.newThread(pool);
            thread.setDaemon(true);
            thread.setName(prefix + count.getAndIncrement());
            return thread;
        }
    }
}
